package com.hfjy.bigdata

import java.util.Properties

import com.hfjy.bigdata.impl.RestClientFactoryImpl
import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.elasticsearch.{ElasticsearchSinkFunction, RequestIndexer}
import org.apache.flink.streaming.connectors.elasticsearch6.{ElasticsearchSink, RestClientFactory}
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011
import org.apache.http.HttpHost
import org.elasticsearch.action.index.IndexRequest
import org.elasticsearch.client.Requests

/**
  * Created by kehailin on 2018-8-16. 
  */
object KafkaFlinkElasticSearch {

    case class WordWithCount(word: String, count: Int)

    def main(args: Array[String]): Unit = {
        val env = StreamExecutionEnvironment.getExecutionEnvironment

        val props = new Properties()
        props.setProperty("bootstrap.servers", "192.168.0.103:9092")
        props.setProperty("group.id", "test1")

        val topic = "thomas_test"

        val kafkaSource = new FlinkKafkaConsumer011[String](topic, new SimpleStringSchema(), props)
        val input: DataStream[String] = env.addSource(kafkaSource)

        val windowCounts: DataStream[WordWithCount] = input.flatMap(w => w.split("\\s"))
            .map(w => WordWithCount(w, 1))
            .keyBy("word")
            .timeWindow(Time.seconds(5))
            .sum("count")

        val httpHosts = new java.util.ArrayList[HttpHost]()
        httpHosts.add(new HttpHost("192.168.0.180", 9200, "http"))  //9200是http port

        val esSinkBuilder = new ElasticsearchSink.Builder[WordWithCount](
            httpHosts,
            new ElasticsearchSinkFunction[WordWithCount] {
                override def process(t: WordWithCount, runtimeContext: RuntimeContext, requestIndexer: RequestIndexer): Unit = {
                    requestIndexer.add(createIndexRequest(t))
                }

                def createIndexRequest(element: WordWithCount): IndexRequest = {
                    val json = new java.util.HashMap[String, Any]()
                    json.put("word", element.word)
                    json.put("count", element.count)

                    Requests.indexRequest()
                        .index("thomas_index4")
                        .`type`("thomas_type4")
                        .source(json)
                }
            }
        )

        esSinkBuilder.setBulkFlushMaxActions(1)
        esSinkBuilder.setRestClientFactory(new RestClientFactoryImpl)
        windowCounts.addSink(esSinkBuilder.build())
        env.execute("kafka to flink to es")
    }
}


